package com.bigdata.hpsk.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  *  词频统计WordCount
  */
object WordCountSpark {

  /**
    * 如果Spark Application运行在本地的话，Driver Program
    *     JVM Process
    */
  def main(args: Array[String]): Unit = {

    /**
      * 从前面spark-shell命令行可知：
      *   Spark 数据分析的程序入口SparkContext，用于读取数据
      */
    // 读取Spark Application的配置信息
    val sparkConf = new SparkConf()
      // 设置SparkApplication名称
      .setAppName("WordCount Spark Application")
      // 设置程序运行的环境，通常情况下，在IDE中开发的时候，设置为local mode，至少是两个Thread
      // 在实际部署的时候通过通过提交应用的命令悍进行设置
      .setMaster("local[2]")
    // 创建SparkContext上下文对象
    val sc = SparkContext.getOrCreate(sparkConf)

    // 设置应用运行日志级别
    sc.setLogLevel("WARN")

/** =================================================================== */
    /**
      * Step 1: read data
      *     SparkContext用于读取数据 -> RDD
      */
    val linesRDD = sc.textFile("/datas/wc.input")   // 从HDFS上文件，所以访问HDFS需要读取配置信息

    /**
      * Step 2: process data
      *     RDD#transformation
      */
    val tupleRDD: RDD[(String, Int)] = linesRDD
      .flatMap(line => {
        line
          .split(" ")  // 对每一行进行分割
          .toList // 转换为List
          .filter(word => word.trim.length > 0) // 过滤单词为空
          .map(word => (word.trim, 1)) // 转换为二元组
      })
    // def reduceByKey(func: (V, V) => V): RDD[(K, V)]
    val wordCountRDD: RDD[(String, Int)] = tupleRDD.reduceByKey((a, b) => a + b)

    // foreach 触发一个job
    wordCountRDD.foreach(println)

    // 对RDD使用多次的话，可以进行缓存
    wordCountRDD.cache()

    // 对统计出来的词频排序
    val sortRDD: RDD[(Int, String)] = wordCountRDD
      // 将word,count 进行调换
      .map(tuple => (tuple._2, tuple._1))
      // 按照key进行排序
      .sortByKey(ascending = false)

    val sortRDD2 = wordCountRDD.sortBy(tuple => tuple._2, ascending = false)

    /**
      * Step 3: write data
      *     将处理的结果数据存储
      *     RDD#action
      */
    println("-------------- WordCount -----------------------")
    // wordCountRDD.foreach 调用的是RDD中的方法，针对RDD操作，在每个Executor中
    wordCountRDD.foreach(tuple => println(tuple._1 + " = " + tuple._2))

    /**
      * RDD#take() 和RDD#top()将数据以数组Array的形式返回给Driver端
      */
    println("-------------- sortByKey -----------------------")
    // 获取前三个值
    sortRDD.take(3).foreach(println)

    println("---------------- sortBy ---------------------")

    sortRDD2.take(3).foreach(tuple => println(tuple._1 + " = " + tuple._2))

    println("----------------- top --------------------")

    wordCountRDD.map(tuple => (tuple._2, tuple._1)).top(3).foreach(println)

    /**
      * 比较map和mapPartitions
      */
    wordCountRDD.map(tuple => {
      println("@@@@@@@@@@@@@@@@@@@@@@@@@@@")
      (tuple._1 + " - " + tuple._2)
    }).count()

    println("-------------------------------------------------")

    wordCountRDD.mapPartitions(iter => {
      println("##############################")
      iter.toList.map(tuple => (tuple._1 + " - " + tuple._2)).toIterator
    }).count()

    // 当不使用内存中RDD时，释放资源
    wordCountRDD.unpersist()




/** =================================================================== */
    // 在开发测试的时候，为了在每个Application页面监控查看应用中Job的运行
    Thread.sleep(1000000)

    // 关闭资源
    sc.stop()
  }

}
